package producer;

import Services.RocketMqProducer.RocketMqProducer;
import Services.RocketMqProducer.RocketMqProducerImpl;
import org.apache.thrift.TException;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TSSLTransportFactory;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TServerTransport;
import org.apache.thrift.transport.TTransportException;

import java.util.Properties;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import config.MqConfig;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import com.aliyun.openservices.ons.api.Message;
import org.apache.log4j.Logger;
import org.ini4j.Ini;
import org.ini4j.Profile.Section;
import java.io.FileReader;
import java.io.IOException;

public class Service {

    private static Logger logger = Logger.getLogger(Service.class);

    public void startServer() throws TTransportException {
        Ini ini = new Ini();
        try {
            //加载配置文件
            ini.load(new FileReader(MqConfig.config_file));
            Section aliyun_section = ini.get(MqConfig.aliyun_section);
            Section producer_section = ini.get(MqConfig.producer_section);

            // rocketMq
            Properties producerProperties = new Properties();
            producerProperties.setProperty(PropertyKeyConst.AccessKey, aliyun_section.get("ACCESS_KEY"));
            producerProperties.setProperty(PropertyKeyConst.SecretKey, aliyun_section.get("SECRET_KEY"));
            producerProperties.setProperty(PropertyKeyConst.NAMESRV_ADDR, aliyun_section.get("NAMESRV_ADDR"));

            Producer producer1 = ONSFactory.createProducer(producerProperties);
            producer1.start(); //普通消息生产者
            OrderProducer producer2 = ONSFactory.createOrderProducer(producerProperties);
            producer2.start(); //顺序消息生产者
            Message message = new Message();
            TProcessor tprocessor = new RocketMqProducer.Processor<RocketMqProducer.Iface>(new RocketMqProducerImpl(message, producer1, producer2));
            Integer port = Integer.valueOf(producer_section.get("SERVER_PORT"));

            // 简单模式
            //TServerSocket serverTransport = new TServerSocket(port);

            // 加证书
            TSSLTransportFactory.TSSLTransportParameters params = new TSSLTransportFactory.TSSLTransportParameters();
            String keyStore = producer_section.get("KeyStore");
            String keyPass = producer_section.get("KeyPass");
            params.setKeyStore(keyStore, keyPass, null, null);
            TServerTransport serverTransport = TSSLTransportFactory.getServerSocket(port, 0, null, params);

            TThreadPoolServer.Args tArgs = new TThreadPoolServer.Args(serverTransport);
            tArgs.processor(tprocessor);
            tArgs.protocolFactory(new TBinaryProtocol.Factory());
            TThreadPoolServer ttps = new TThreadPoolServer(tArgs);
            ttps.serve();
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws TException {
        System.out.println("Producer start success.");
        logger.info("producer service start");
        try {
            Service hss = new Service();
            hss.startServer();
        }catch (TTransportException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
}
